fix: report Input column when native Iceberg scan is enabled or native shuffle is enabled - #5265

Thanks for tracking this one down. The diagnosis looks right to me. I traced the shuffle path as well, and CometShuffleExchangeExec.nativeChildContext matches on child: CometNativeExec, which CometIcebergNativeScanExec satisfies directly, so spark.table(t).repartition(...) really was producing hasScanInput = false before this change. The two reporting sites you identified are the right ones.

A few things I would like your thoughts on.

The comment overstates what CSV contributes

CometCsvNativeScanExec has no metrics override, so it inherits CometNativeExec.metrics, which is CometMetricNode.baselineMetrics and only carries output_rows and elapsed_compute. There is no bytes_scanned. The native side matches, native/core/src/execution/operators/csv_scan.rs registers no metrics at all.

So CSV scans still report nothing after this change. The scanLeaves.filter(_.metrics.contains("bytes_scanned")) guard in reportScanInputMetrics makes that a safe no-op, which the second half of your comment does say. But naming CometCsvNativeScanExec in the list of scans that "can contribute bytes_scanned / output_rows" reads like it is fixed here. Could we drop CSV from that list, or note that it has no bytes_scanned metric yet? The PR title and description carry the same implication.

The gate now depends on a filter in another class

The comment justifies the broad match by saying reportScanInputMetrics self-filters on bytes_scanned. That is accurate today, but it makes the correctness of hasScanInput a function of an implementation detail in CometMetricNode, with nothing linking the two. What do you think about expressing the condition directly instead?

hasScanInput = sparkPlans.exists(p =>
  p.isInstanceOf[CometLeafExec] && p.metrics.contains("bytes_scanned"))

That says what it means without the cross-class reasoning. It still picks up contrib scans without a compile-time reference, which was the original motivation for matching on the base class. It also avoids registering a task-completion listener for plans where there is nothing to report, and it would let the comment shrink to a line or two.

Test duplication

These are good tests. Asserting the plan shape before the metrics is the right call, and the note about pinning COMET_SHUFFLE_MODE rather than trusting auto is appreciated.

Between these two and the existing SELECT * test at line 3580 there are now three copies of the catalog config, the table DDL, the range(10000).repartition(5) load, the listener, and the drain-clear-run-drain sequence. Would it be worth pulling that into a helper that takes a table name and a body? It would shrink the diff here and make the next metrics test much cheaper to add.

Listener asymmetry between the two new tests

The listener in the fused-block test filters on if (im.bytesRead > 0) before recording, while the shuffle test records unconditionally. The unconditional version is stronger, because a filtered listener cannot tell "every task reported zero" apart from "no task ran". Both leave the buffer empty. I realise the first one mirrors the existing test above it, but could both new ones drop the filter?

A follow-up, not something for this PR

reportScanInputMetrics uses setBytesRead / setRecordsRead rather than the incrementing variants. So if a fallback Spark scan reaches a native block through CometSparkToColumnarExec alongside a native scan, Comet's completion listener overwrites whatever Spark's FileScanRDD accumulated for that task. That is pre-existing behaviour for CometNativeScanExec and not introduced here, but this change widens the set of plans that can hit it. Could you file a tracking issue so it does not get lost? Happy to link it from here.

CI

The head commit does not have any check runs yet. The previous commit was green, 28 success and 2 skipped, and that run did include the [scans] bucket that carries CometIcebergNativeSuite on Spark 4.1 and 4.2. It did not show [scans] for Spark 3.4 or 3.5 though. Worth letting a full run finish on the current head before merging.

I used an LLM to help work through this review. Please treat the points above as a starting point rather than a checklist. You are much closer to this code than I am, so use your judgement on which ones are worth acting on and push back or just reply in the thread on any you disagree with.

Note on this review: this was generated by an LLM (Claude Code) at my request while I worked through a review backlog. I have not verified the individual findings myself. Please treat everything below as suggestions to evaluate rather than as authoritative review feedback, and push back on anything that is wrong or already handled.

The diagnosis is exactly right, and matching on CometLeafExec rather than CometNativeScanExec is the minimal fix. The comment explaining why the self-filtering in reportScanInputMetrics makes it safe for leaves that do not track bytes_scanned is the part I wanted to see, and it is there.

Three things.

Is there a regression test for the Parquet case?

The change widens the condition, so the risk is double counting rather than under counting: a leaf that both reports its own input metrics in doExecuteColumnar and now also gets reported through CometNativeExec.executeColumnarWithContext. The new test covers the fused Iceberg shape. Is there an existing test that pins the non-fused Parquet shape, so we would notice if a leaf started being counted twice? If not, an assertion that bytesRead equals roughly the file size rather than twice it would be worth adding.

CSV and contributed scans

The comment names CometCsvNativeScanExec and contrib scans as beneficiaries. Do they actually populate bytes_scanned? If they do not today, the change is a no-op for them and that is fine, but the comment reads as though they are fixed by this too. Worth being precise about which scans this actually turns on.

The test is 237 lines for a one-line fix

That is not a criticism of the test, which is careful and well commented, especially the fusingParents assertion that stops it silently degrading into a duplicate of the SELECT * test. But CometIcebergNativeSuite only runs when Iceberg is on the classpath and it is already a long suite. Could the same property be asserted more cheaply against a Parquet scan plus a projection, with the Iceberg case reduced to confirming the scan type is recognized? If the Iceberg-specific setup is load-bearing then keep it as is, but it is worth asking whether the coverage has to cost this much wall clock.